package com.fxkj.cloud.drms.ui.mqtt.handler.receive;

import com.fxkj.cloud.dbs.common.constants.DeviceStatusEnum;
import com.fxkj.cloud.drms.application.DeviceApplicationService;
import com.fxkj.cloud.dbs.common.model.kv.BasicTsKvEntry;
import com.fxkj.cloud.dbs.common.model.kv.KvEntry;
import com.fxkj.cloud.dbs.common.model.kv.TsKvEntry;
import com.fxkj.cloud.dbs.common.model.kv.util.TelemetryJacksonUtil;
import com.fxkj.cloud.drms.domain.ts.service.TelemetryService;
import com.fxkj.cloud.drms.infra.mq.mqtt.MqttConstants;
import com.fxkj.cloud.drms.ui.mqtt.handler.AbstractReceiveHandler;
import com.fxkj.cloud.drms.ui.mqtt.model.DeviceMsg;
import com.google.common.util.concurrent.FutureCallback;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

/**
 * <p>description: </p>
 * @author ww
 * @date 2025/3/17 11:14
 */
@Component(MqttConstants.Topic.Receive.DEVICE_MSG_RECEIVER_SERVICE_NAME)
@Slf4j
public class DeviceMsgReceiverHandler extends AbstractReceiveHandler<DeviceMsg> {
    private final DeviceApplicationService deviceApplicationService;
    private final TelemetryService telemetryService;

    public DeviceMsgReceiverHandler(DeviceApplicationService deviceApplicationService, TelemetryService telemetryService) {
        this.deviceApplicationService = deviceApplicationService;
        this.telemetryService = telemetryService;
    }

    @Override
    protected void doHandler(String topic, DeviceMsg data, String... topicParams) throws Exception {
        log.info("[DeviceMsgReceiverHandler] -| doHandler -| topicParams: [{}], topic: {}, data: {}", topicParams, topic, data);
        String serialNo = topicParams[0];
        Map<Long, List<KvEntry>> tsKvMap = data.getValues();
        List<TsKvEntry> tsKvEntryList = new ArrayList<>();
        for (Map.Entry<Long, List<KvEntry>> tsKvEntry : tsKvMap.entrySet()) {
            for (KvEntry kvEntry : tsKvEntry.getValue()) {
                tsKvEntryList.add(new BasicTsKvEntry(tsKvEntry.getKey(), kvEntry));
            }
        }
        telemetryService.saveAndNotify(serialNo, tsKvEntryList, new FutureCallback<Void>() {
            @Override
            public void onSuccess(Void result) {
                log.debug("数据处理成功");
            }

            @Override
            public void onFailure(Throwable t) {
                log.warn("数据处理失败");
            }
        });
        deviceApplicationService.deviceStatusChange(serialNo, DeviceStatusEnum.WORKING);
    }

    @Override
    protected Class<DeviceMsg> getClazz() {
        return DeviceMsg.class;
    }

    @Override
    protected DeviceMsg parseData(byte[] msgBytes) {
        String src = new String(msgBytes);
        JsonElement jsonElement = JsonParser.parseString(src);
        JsonObject jsonObject = jsonElement.getAsJsonObject();
        long ts = jsonObject.get("ts").getAsLong();
        JsonElement requestIdJsonElement = jsonObject.get("requestId");
        UUID requestId = null;
        if (null != requestIdJsonElement && !requestIdJsonElement.isJsonNull()) {
            requestId = UUID.fromString(requestIdJsonElement.getAsString());
        }
        JsonElement valuesJsonElement = jsonObject.get("values");
        long now = System.currentTimeMillis();
        Map<Long, List<KvEntry>> values = TelemetryJacksonUtil.convertToTelemetry(valuesJsonElement, now);
        return new DeviceMsg(ts, values, requestId);
    }
}
